package 多线程.实战项目测试;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.Map;
import java.util.concurrent.*;

/**
 * 做这个多线程异步任务，主要是因为我们有很多永动的异步任务，什么是永动呢？就是任务跑起来后，需要⼀直跑
 * 下去，⽐如消息Push任务，因为⼀直有消息过来，所以需要⼀直去消费DB中的未推送消息，就需要整⼀个Push的
 * 永动异步任务。
 * 我们的需求其实不难，简单总结⼀下：
 * 1. 能同时执⾏多个永动的异步任务；
 * 2. 每个异步任务，⽀持开多个线程去消费这个任务的数据；
 * 3. ⽀持永动异步任务的优雅关闭，即关闭后，需要把所有的数据消费完毕后，再关闭。
 * 完成上⾯的需求，需要注意⼏个点：
 * 1. 每个永动任务，可以开⼀个线程去执⾏；
 * 2. 每个⼦任务，因为需要⽀持并发，需要⽤线程池控制；
 * 3. 永动任务的关闭，需要通知⼦任务的并发线程，并⽀持永动任务和并发⼦任务的优雅关闭。
 *
 *
 * 对于⼦任务，需要⽀持并发，如果每个并发都开⼀个线程，⽤完就关闭，对资源消耗太⼤，所以引⼊线程池：
 */

public class TaskProcessUtil {
    // 每个任务，都有⾃⼰单独的线程池
    private static Map<String, ExecutorService> executors = new ConcurrentHashMap<>();

    // 初始化⼀个线程池
    private static ExecutorService init(String poolName, int poolSize) {
        return new ThreadPoolExecutor(poolSize, poolSize,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactoryBuilder().setNameFormat("Pool-" +
                        poolName).setDaemon(false).build(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    // 获取线程池
    public static ExecutorService getOrInitExecutors(String poolName, int poolSize) {
        ExecutorService executorService = executors.get(poolName);
        if (null == executorService) {
            synchronized (TaskProcessUtil.class) {
                executorService = executors.get(poolName);
                if (null == executorService) {
                    executorService = init(poolName, poolSize);
                    executors.put(poolName, executorService);
                }
            }
        }
        return executorService;
    }

    // 回收线程资源
    public static void releaseExecutors(String poolName) {
        ExecutorService executorService = executors.remove(poolName);
        if (executorService != null) {
            executorService.shutdown();
        }
    }
}